package com.ylkj.bus.app.report;

import com.alibaba.fastjson2.JSONObject;
import com.ylkj.bus.MessageContent;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;

import java.lang.ref.SoftReference;
import java.sql.Connection;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;


/**
 * @Author: wengzx
 * @Date: 2025/3/17
 * @Desc:
 */
@Slf4j
@RequiredArgsConstructor
public class ReportAppHandler extends ProcessFunction<MessageContent, MessageContent> {

    private static final String T_IOTP_APP_SEARCH_SQL = "SELECT id,app_name as appName FROM {0} where app_desc!=\"容器预置APP\" limit ? offset ?";

    private final RdsConfig rdsConfig;

    private final OutputTag<List<IotmoAppDO>> outTagRds1;
    private final OutputTag<List<ContainerAppDO>> outTagRds2;
    private final String rds4TableNameWithJoin;


    // appName,id
    private final Map<String, String> iotpAppcaches = new HashMap<>();

    private RdsUtils rdsUtils;
    private String rds5SQL;


    public static final String TYPE = "type";
    /**
     * 应用状态上报
     */
    public static final String STATUS = "REP_APP_STATUS";

    /**
     * topic: /v1/${edgeId}/app/data
     * type: REP_APP_STATUS
     * 服务状态，running 或 stopped
     */
    public static final String APP_SERVER_STATUS_RUNNING = "running";

    /**
     * topic: /v1/${edgeId}/app/data
     * type: REP_APP_STATUS
     * 服务使能状态，yes 或 no
     */
    public static final String APP_SERVER_ENABLE_YES = "yes";


    /**
     * app服务状态 SERVICE_STATUS  字典表  01:启动,  02:停止
     */
    public static final String SERVICE_STATUS_01 = "01";
    public static final String SERVICE_STATUS_02 = "02";

    /**
     * app使能状态 ENABLED_STATUS  字典表  01:使能, 02:去使能
     */
    public static final String ENABLED_STATUS_01 = "01";
    public static final String ENABLED_STATUS_02 = "02";

    /**
     * app操作状态 OPTION_STATUS  字典表  01:指令下发成功, 04:升级中,05:升级失败,06:升级成功,07:安装成功,08:安装失败,09:已卸载
     */
    public static final String OPTION_STATUS_07 = "07";

    /**
     * source 1:平台新增 2：终端预置
     */
    public static final int CONTAINER_APP_SOURCE_PRESET = 2;

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        rds5SQL=MessageFormat.format(T_IOTP_APP_SEARCH_SQL,rds4TableNameWithJoin);
        initIotpAppCache();
    }
    @Override
    public void processElement(MessageContent mqttMsg, ProcessFunction<MessageContent, MessageContent>.Context ctx, Collector<MessageContent> out) throws Exception {
        JSONObject body = JSONObject.parseObject(mqttMsg.getBody());
        //获取msg数据类型
        String type = body.getString(TYPE);
        if(!Objects.equals(type,STATUS)){
            if (log.isDebugEnabled()) {
                log.debug(" 类型错误 {} {} {}", type, STATUS, body);
            }
            return;
        }
        String topic = mqttMsg.getTopic(); //主题格式:/v1/{}/app/data
        String deviceCode = topic.split("/")[2];
        try {

            reportAppStatus(type, body, deviceCode, topic, ctx);
        } catch (Exception e) {
            log.error("原始数据信息：{}", mqttMsg);
            log.error(e.getMessage(), e);
        }
    }

    /**
     * APP状态上报
     * @param type
     * @param body
     * @param deviceCode
     * @param topic
     * @param ctx
     */
    private void reportAppStatus(String type, JSONObject body, String deviceCode, String topic, ProcessFunction<MessageContent, MessageContent>.Context ctx) throws SQLException {
        ReportAppStatus reportAppStatus = body.to(ReportAppStatus.class);
        ReportAppStatusParam param = reportAppStatus.getParam();
        String containerName = param.getContainer();//容器名称
        List<StatusParamApp> paramAppList = param.getApps();
        if(CollectionUtils.isEmpty(paramAppList)){
            if (log.isDebugEnabled()) {
                log.debug("topic:{} 应用状态上报数据为空 {} {} {}", topic, type, STATUS, body);
            }
            return;
        }
        addMonitorAppInfo(deviceCode, containerName, paramAppList, ctx);
    }


    /**
     * 插入应用监控数据
     *
     * @param deviceCode
     * @param containerName
     * @param paramAppList
     */
    private void addMonitorAppInfo(String deviceCode, String containerName, List<StatusParamApp> paramAppList,  ProcessFunction<MessageContent, MessageContent>.Context ctx) {


        List<IotmoAppDO> iotmoAppDOS = new ArrayList<>(paramAppList.size());
        List<ContainerAppDO> containerAppDOS = new ArrayList<>(paramAppList.size());
        for (StatusParamApp statusParamApp : paramAppList) {
            StatusParamAppProcessVO statusParamAppProcessVO = statusParamApp.getProcess().get(0);
            //更新app数据上报表
            IotmoAppDO iotmoAppDO = new IotmoAppDO();
            //获取app名称
            String appName = statusParamApp.getApp();
            //String appVoId = UUID.randomUUID().toString();
            String appVoId = String.format("%s_%s_%s", deviceCode, containerName, appName);
            iotmoAppDO.setId(appVoId);
            iotmoAppDO.setDeviceIdentifier(deviceCode);
            iotmoAppDO.setAppName(appName);
            iotmoAppDO.setAppVersion(statusParamApp.getVersion());
            iotmoAppDO.setContainerName(containerName);
            iotmoAppDO.setLifeTime(statusParamApp.getLifeTime());
            iotmoAppDO.setServiceStatus(APP_SERVER_STATUS_RUNNING.equals(statusParamAppProcessVO.getSrvStatus()) ? SERVICE_STATUS_01 : SERVICE_STATUS_02);
            iotmoAppDO.setEnableStatus(APP_SERVER_ENABLE_YES.equals(statusParamAppProcessVO.getSrvEnable()) ? ENABLED_STATUS_01 : ENABLED_STATUS_02);

            ContainerAppDO containerAppDO = new ContainerAppDO();
            //获取容器id
            String containerId = String.format("%s_%s", deviceCode, containerName);
            //containerAppDO.setId(UUID.randomUUID().toString());
            containerAppDO.setId(appVoId);
            containerAppDO.setContainerId(containerId);
            containerAppDO.setAppId(iotpAppcaches.get(statusParamApp.getApp()));
            containerAppDO.setOptionStatus(OPTION_STATUS_07);
            containerAppDO.setSerivceStatus(APP_SERVER_STATUS_RUNNING.equals(statusParamAppProcessVO.getSrvStatus()) ? SERVICE_STATUS_01 : SERVICE_STATUS_02);
            containerAppDO.setEnableStatus(APP_SERVER_ENABLE_YES.equals(statusParamAppProcessVO.getSrvEnable()) ? ENABLED_STATUS_01 : ENABLED_STATUS_02);
            containerAppDO.setBelongId(appVoId);
            containerAppDO.setSource(CONTAINER_APP_SOURCE_PRESET);
            containerAppDO.setAppHash(statusParamApp.getAppHash());
            containerAppDO.setLifeTime(statusParamApp.getLifeTime());
            iotmoAppDOS.add(iotmoAppDO);
            containerAppDOS.add(containerAppDO);
            if (log.isDebugEnabled()){
                log.debug("iotmoAppDO:{} \n containerAppDO:{}",iotmoAppDO,containerAppDO);
            }
        }
        ctx.output(outTagRds1, iotmoAppDOS);
        ctx.output(outTagRds2, containerAppDOS);

    }

    private void initIotpAppCache() throws SQLException {
        try (Connection connection = rdsUtils.getConnection()) {
            //创建 QueryRunner 对象
            QueryRunner queryRunner = new QueryRunner();
            //设置分页参数
            int pageSize = 3000;
            int currentPage = 1;
            while (true){
                log.info("加载{}缓存 {} {}", rds4TableNameWithJoin, currentPage, pageSize);
                int offset = (currentPage - 1) * pageSize;
                List<IotmoAppDO> query = queryRunner.query(connection, rds5SQL, new BeanListHandler<>(IotmoAppDO.class), pageSize, offset);
                if (query == null || query.isEmpty()) {
                    break;
                }
                for (IotmoAppDO iotmoAppDO : query) {
                    iotpAppcaches.put(iotmoAppDO.getAppName(),iotmoAppDO.getId());
                }
                currentPage++;
            }
            log.info("{} 缓存加载完成 {}",rds4TableNameWithJoin,iotpAppcaches.size() );
        }catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private static String getCacheKey(String deviceCode, String containerName) {
        return String.join(" ", deviceCode, containerName);
    }

    @Override
    public void close() throws Exception {
        rdsUtils.close();
    }
}
